package com.vertx.kafka;

import io.vertx.core.Vertx;
import io.vertx.kafka.client.consumer.KafkaConsumer;
import io.vertx.kafka.client.consumer.KafkaConsumerRecord;

import java.time.Duration;
import java.util.HashMap;

public class KafkaConsumerPeriodicHandler {

    public static void main(String[] args) {

        var config = new HashMap<String, String>();

        config.put("bootstrap.servers", "172.18.30.14:9092");
        config.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        config.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        config.put("group.id", "zhengxing-sync-test.zhengxing.statistic_device-ceshi006");
        config.put("auto.offset.reset", "earliest");
        config.put("enable.auto.commit", "false");

        var vertx = Vertx.vertx();
        KafkaConsumer<String,String> consumer = KafkaConsumer.create(vertx, config);

        var topic = "zhengxing-sync-test.zhengxing.statistic_device";

        // 订阅topic
        consumer.subscribe(topic)
                .onSuccess(h->{
                    System.out.println("subscribe success");
                })
                .onFailure(e->{
                    System.out.println("subscribe error " + e);
                });
        // 分区处理
        consumer.partitionsAssignedHandler(h->{
            System.out.println("partition assigned");
            h.forEach(tp->{
                System.out.println(tp.getPartition());
            });
        })
                .partitionsRevokedHandler(h->{
                    System.out.println("partition revoked");
                    h.forEach(tp->{
                        System.out.println(tp.getPartition());
                    });
                });

        // 设置了一个定时器 定时去手动消费数据
        vertx.setPeriodic(1000, timerId ->{
                    System.out.println("*******************");
                consumer
                        // pool(timeout)  阻塞获取数据的时间
                        .poll(Duration.ofMillis(100))
                        .onSuccess(records -> {
                            System.out.println("-------------------------------" + records.size());
                            for (int i = 0; i < records.size(); i++) {
                                KafkaConsumerRecord<String, String> record = records.recordAt(i);
                                System.out.println("key=" + record.key() + ",value=" + record.value() +
                                        ",partition=" + record.partition() + ",offset=" + record.offset());
                            }
                        })
                        .onFailure(cause -> {
                            System.out.println("Something went wrong when polling " + cause.toString());
                            cause.printStackTrace();

                            // Stop polling if something went wrong
                            vertx.cancelTimer(timerId);
                        });
        }
        );

    }

}
